/*
 * contrib/hstore/hstore_compat.c
 *
 * 关于旧/新 hstore 格式歧义的说明。
 *
 * 有三种格式需要考虑：
 * 1) 旧的 contrib/hstore（称为 hstore-old）
 * 2) 预发布的 pgfoundry hstore
 * 3) 新的 contrib/hstore
 *
 * （2）和（3）是相同的，除了 HS_FLAG_NEWVERSION
 * 位，在（3）中设置，但在（2）中不设置。
 *
 * 已经是格式（3）或明确属于格式（2）的值
 * 由 hstoreUpgrade() 中的第一个“立即返回”测试处理。
 *
 * 强调一点：只有在我们从旧的预发布 pgfoundry hstore-new 进行
 * 某种就地迁移时，可能出现模糊值，我们才会到这里；而且我们明确
 * 不支持这种情况，而不先修复任何可能的填充值。这里的大部分代码
 * 是严重的多余，但是性能损失并不严重（尤其是与我们无论如何都要
 * 执行的 palloc() 相比），而且安全和稳妥的有效性检查提供了一些
 * 安心。（如果由于某种原因我们得到一个在旧代码中可以工作的值，
 * 但是会被转换代码搞砸，有效性检查会首先使其失败，因此我们会
 * 得到一个错误，而不是坏数据。）
 *
 * 还要注意，空 hstore 在（2）和（3）中是相同的，因此
 * 对它们有一些特殊情况。
 *
 * 我们区分格式（2）和（3）如下（但要注意，有些边缘情况
 * 我们无法判断；请参见 hstoreUpgrade 中的注释）：
 *
 * 首先，由于至少必须有一个条目，我们查看
 * 位的排列方式。新格式看起来像：
 *
 * 10kkkkkkkkkkkkkkkkkkkkkkkkkkkkkk  （k..k = keylen）
 * 0nvvvvvvvvvvvvvvvvvvvvvvvvvvvvvv  （v..v = keylen+vallen）
 *
 * 旧格式看起来像其中一个，具体取决于字节序
 * 和位域布局：（k..k = keylen，v..v = vallen，p..p = pos，
 * n = isnull）
 *
 * kkkkkkkkkkkkkkkkvvvvvvvvvvvvvvvv
 * nppppppppppppppppppppppppppppppp
 *
 * kkkkkkkkkkkkkkkkvvvvvvvvvvvvvvvv
 * pppppppppppppppppppppppppppppppn
 *
 * vvvvvvvvvvvvvvvvkkkkkkkkkkkkkkkk
 * nppppppppppppppppppppppppppppppp
 *
 * vvvvvvvvvvvvvvvvkkkkkkkkkkkkkkkk
 * pppppppppppppppppppppppppppppppn   （通常的 i386 格式）
 *
 * 如果条目是旧格式，则对于第一个条目，“pos”必须为 0。
 * 我们显然可以看到，只有当 keylen 或 vallen 大于 32768 时
 * 才会出现任何歧义（这就是为什么在 hstore.h 中将长度小于此的
 * 快速处理）。由于“pos”==0，新的格式解释中的“v”字段只能为
 * 0 或 1，这限制了旧格式的 k 和 v 字段中的所有但三个位。
 * 但是，除了所有这些之外，由 keylen 和 vallen 隐含的数据长度
 * 必须适合 varlena 大小。因此，只有一个条目的 hstore 的
 * 唯一歧义边缘情况发生在具有超出 (~32k) 填充的新格式条目和
 * 旧格式条目之间。但是根据我们编译时的方式，我们知道在那种情况下
 * 使用哪种格式，因此不会发生实际的数据损坏。
 *
 * 如果条目超过一个，要求键长度不减少，位置连续增加，
 * 并且数据的末尾不得超出 varlena 本身的末尾，这在几乎所有其他情况下
 * 都能消除歧义。如果旧格式值有大量超出填充量和刚好合适的键
 * 大小模式，则可能会出现一小部分歧义情况，但这些也根据我们
 * 编译时的方式处理。
 *
 * 另提供未文档化的函数 hstore_version_diag 供测试之用。
 */
#include "postgres.h"


#include "hstore.h"

/*
 * 这是用于旧版 contrib/hstore 实现中条目的结构。注意，这与新条目的大小相同
 * （每个键/值对两个 32 位字），并且头部相同，因此 ARRPTR、STRPTR、CALCDATASIZE
 * 等的旧版和新版是兼容的。
 *
 * 如果上述声明在某些奇怪的平台上不成立，我们会有点麻烦（参见 hstoreValidOldFormat 中的 StaticAssertStmt）。
 */
typedef struct
{
	uint16		keylen;
	uint16		vallen;
	uint32
				valisnull:1,
				pos:31;
} HOldEntry;

static int	fc_hstoreValidNewFormat(HStore *fc_hs);
static int	fc_hstoreValidOldFormat(HStore *fc_hs);


/*
 * 新格式 hstore 的有效性测试。
 *	0 = 不有效
 *	1 = 有效但长度有“冗余”
 *	2 = 完全有效
 */
static int fc_hstoreValidNewFormat(HStore *fc_hs)
{
	int			fc_count = HS_COUNT(fc_hs);
	HEntry	   *fc_entries = ARRPTR(fc_hs);
	int			fc_buflen = (fc_count) ? HSE_ENDPOS(fc_entries[2 * (fc_count) - 1]) : 0;
	int			fc_vsize = CALCDATASIZE(fc_count, fc_buflen);
	int			fc_i;

	if (fc_hs->size_ & HS_FLAG_NEWVERSION)
		return 2;

	if (fc_count == 0)
		return 2;

	if (!HSE_ISFIRST(fc_entries[0]))
		return 0;

	if (fc_vsize > VARSIZE(fc_hs))
		return 0;

	/* 条目位置必须是非递减的 */

	for (fc_i = 1; fc_i < 2 * fc_count; ++fc_i)
	{
		if (HSE_ISFIRST(fc_entries[fc_i])
			|| (HSE_ENDPOS(fc_entries[fc_i]) < HSE_ENDPOS(fc_entries[fc_i - 1])))
			return 0;
	}

	/* 键长度必须是非递减的，并且键不能为空 */

	for (fc_i = 1; fc_i < fc_count; ++fc_i)
	{
		if (HSTORE_KEYLEN(fc_entries, fc_i) < HSTORE_KEYLEN(fc_entries, fc_i - 1))
			return 0;
		if (HSE_ISNULL(fc_entries[2 * fc_i]))
			return 0;
	}

	if (fc_vsize != VARSIZE(fc_hs))
		return 1;

	return 2;
}

/*
 * 旧格式 hstore 的有效性测试。
 *	0 = 不有效
 *	1 = 有效但长度有“冗余”
 *	2 = 完全有效
 */
static int fc_hstoreValidOldFormat(HStore *fc_hs)
{
	int			fc_count = fc_hs->size_;
	HOldEntry  *fc_entries = (HOldEntry *) ARRPTR(fc_hs);
	int			fc_vsize;
	int			fc_lastpos = 0;
	int			fc_i;

	if (fc_hs->size_ & HS_FLAG_NEWVERSION)
		return 0;

	/* 新格式使用 HEntry 作为键，另一个作为值 */
	StaticAssertStmt(sizeof(HOldEntry) == 2 * sizeof(HEntry),
					 "old hstore format is not upward-compatible");

	if (fc_count == 0)
		return 2;

	if (fc_count > 0xFFFFFFF)
		return 0;

	if (CALCDATASIZE(fc_count, 0) > VARSIZE(fc_hs))
		return 0;

	if (fc_entries[0].pos != 0)
		return 0;

	/* 键长度必须是非递减的 */

	for (fc_i = 1; fc_i < fc_count; ++fc_i)
	{
		if (fc_entries[fc_i].keylen < fc_entries[fc_i - 1].keylen)
			return 0;
	}

	/*
	 * 条目位置必须严格递增，除了第一个条目
	 * （可以是 ""=>""，因此长度为零）；并且所有条目必须
	 * 正确相邻
	 */

	for (fc_i = 0; fc_i < fc_count; ++fc_i)
	{
		if (fc_entries[fc_i].pos != fc_lastpos)
			return 0;
		fc_lastpos += (fc_entries[fc_i].keylen
					+ ((fc_entries[fc_i].valisnull) ? 0 : fc_entries[fc_i].vallen));
	}

	fc_vsize = CALCDATASIZE(fc_count, fc_lastpos);

	if (fc_vsize > VARSIZE(fc_hs))
		return 0;

	if (fc_vsize != VARSIZE(fc_hs))
		return 1;

	return 2;
}


/*
 * hstoreUpgrade: PG_DETOAST_DATUM 加上对旧 hstore 转换的支持
 */
HStore *
hstoreUpgrade(Datum fc_orig)
{
	HStore	   *fc_hs = (HStore *) PG_DETOAST_DATUM(fc_orig);
	int			fc_valid_new;
	int			fc_valid_old;

	/* 如果不需要转换，立即返回 */
	if (fc_hs->size_ & HS_FLAG_NEWVERSION)
		return fc_hs;

	/* 我们有可写的副本吗？如果没有，创建一个。 */
	if ((void *) fc_hs == (void *) DatumGetPointer(fc_orig))
		fc_hs = (HStore *) PG_DETOAST_DATUM_COPY(fc_orig);

	if (fc_hs->size_ == 0 ||
		(VARSIZE(fc_hs) < 32768 && HSE_ISFIRST((ARRPTR(fc_hs)[0]))))
	{
		HS_SETCOUNT(fc_hs, HS_COUNT(fc_hs));
		HS_FIXSIZE(fc_hs, HS_COUNT(fc_hs));
		return fc_hs;
	}

	fc_valid_new = fc_hstoreValidNewFormat(fc_hs);
	fc_valid_old = fc_hstoreValidOldFormat(fc_hs);

	if (!fc_valid_old || fc_hs->size_ == 0)
	{
		if (fc_valid_new)
		{
			/*
			 * 强制“新版本”标志和正确的 varlena 长度。
			 */
			HS_SETCOUNT(fc_hs, HS_COUNT(fc_hs));
			HS_FIXSIZE(fc_hs, HS_COUNT(fc_hs));
			return fc_hs;
		}
		else
		{
			elog(ERROR, "invalid hstore value found");
		}
	}

	/*
	 * 这是一个棘手的边缘情况。只有在某些极端情况下
	 * 可能（hstore 的末尾必须有大量浪费的填充空间）。
	 * 但是，只有在这种情况下，“新” hstore 值才能到达这里，
	 * 是因为我们正在从 hstore-new 的预发布版本
	 * 进行就地升级（而不是 contrib/hstore），因此我们基于以下假设工作：
	 * 1. 如果你从旧的 contrib/hstore 移动到 hstore-new，
	 * 你需要首先修复任何潜在冲突，例如，通过在升级前对所有 hstore 列运行 ALTER TABLE ... USING col::text::hstore;
	 * 2. 如果你从旧的 contrib/hstore 移动到新的 contrib/hstore，
	 * 则这里的“新”值是不可能的
	 * 3. 如果你从预发布的 hstore-new 移动到 hstore-new，
	 * 则这里的“旧”值是不可能的
	 * 4. 如果你从预发布的 hstore-new 移动到新的 contrib/hstore，
	 * 你不是在进行就地升级，因此没有问题
	 * 所以总的来说，这一切的结果是，
	 * 如果我们被构建为 hstore-new，我们可以将所有边缘情况视为“新”，
	 * 如果我们被构建为 contrib/hstore，则视为“旧”。
	 *
	 * XXX 一旦经过测试，这条警告可能可以降级为 DEBUG1。
	 * 但现在，如果有人能在非人为设置中实际遇到这种情况，将会非常有用。
	 */

	if (fc_valid_new)
	{
#ifdef HSTORE_IS_HSTORE_NEW
		elog(WARNING, "ambiguous hstore value resolved as hstore-new");

		/*
		 * 强制“新版本”标志和正确的 varlena 长度。
		 */
		HS_SETCOUNT(fc_hs, HS_COUNT(fc_hs));
		HS_FIXSIZE(fc_hs, HS_COUNT(fc_hs));
		return fc_hs;
#else
		elog(WARNING, "ambiguous hstore value resolved as hstore-old");
#endif
	}

	/*
	 * 必须具有旧式值。就地覆盖为新式值。
	 */
	{
		int			fc_count = fc_hs->size_;
		HEntry	   *fc_new_entries = ARRPTR(fc_hs);
		HOldEntry  *fc_old_entries = (HOldEntry *) ARRPTR(fc_hs);
		int			fc_i;

		for (fc_i = 0; fc_i < fc_count; ++fc_i)
		{
			uint32		fc_pos = fc_old_entries[fc_i].pos;
			uint32		fc_keylen = fc_old_entries[fc_i].keylen;
			uint32		fc_vallen = fc_old_entries[fc_i].vallen;
			bool		fc_isnull = fc_old_entries[fc_i].valisnull;

			if (fc_isnull)
				fc_vallen = 0;

			fc_new_entries[2 * fc_i].entry = (fc_pos + fc_keylen) & HENTRY_POSMASK;
			fc_new_entries[2 * fc_i + 1].entry = (((fc_pos + fc_keylen + fc_vallen) & HENTRY_POSMASK)
											| ((fc_isnull) ? HENTRY_ISNULL : 0));
		}

		if (fc_count)
			fc_new_entries[0].entry |= HENTRY_ISFIRST;
		HS_SETCOUNT(fc_hs, fc_count);
		HS_FIXSIZE(fc_hs, fc_count);
	}

	return fc_hs;
}


PG_FUNCTION_INFO_V1(hstore_version_diag);
Datum hstore_version_diag(PG_FUNCTION_ARGS)
{
	HStore	   *fc_hs = (HStore *) PG_DETOAST_DATUM(PG_GETARG_DATUM(0));
	int			fc_valid_new = fc_hstoreValidNewFormat(fc_hs);
	int			fc_valid_old = fc_hstoreValidOldFormat(fc_hs);

	PG_RETURN_INT32(fc_valid_old * 10 + fc_valid_new);
}
